//package com.atguigu.app.dwd.db;
//
//import com.atguigu.utils.KafkaUtil;
//import com.atguigu.utils.MysqlUtil;
//import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
//import org.apache.flink.table.api.Table;
//import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
//
////交易域支付成功事实表
///*
//1.支付主表payment_info(topic_db) 支付成功的状态是1602
//2.interval join(要求两张表都要设置事件时间和watermark) 下单事务事实表 dwd_trade_order_detail 状态保存15min
//3.lookup join（要求主流建表时要设置处理时间字段）字典表 base_dic 退化支付类型
//
////todo 3.创建动态表topic_db连接kafka topic_db
////todo 4.过滤出支付表payment_info支付成功的内容
////todo 5.创建动态表连接kafka的下单事务事实表dwd_trade_order_detail
////todo 6.创建动态表base_dic读取mysql base_dic表
////todo 7.三表关联
////todo 8.创建动态表trade_order_detail连接kafka dwd_trade_order_detail主题
////todo 8.创建动态表payment_detail_suc连接kafka dwd_trade_payment_detail_suc
////todo 9.将关联后的结果查询出来写到动态表payment_detail_suc里
// */
//public class DwdTradePayDetailSuc {
//    public static void main(String[] args) {
//        //todo 1.准备环境
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
//
//        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
//
//        //todo 2.设置检查点和状态后端
//        //        // 需要从Checkpoint或者Savepoint启动程序 需启动hdfs
////        //开启Checkpoint,每隔5秒钟做一次CK  ,并指定CK的一致性语义
////        env.enableCheckpointing(3000L, CheckpointingMode.EXACTLY_ONCE);
////        //设置超时时间为 1 分钟
////        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
////        //设置两次重启的最小时间间隔
////        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
////        //设置任务关闭的时候保留最后一次 CK 数据
////        env.getCheckpointConfig().enableExternalizedCheckpoints(
////                CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
////        //指定从 CK 自动重启策略
////        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(1L), Time.minutes(1L)));
////        //设置状态后端
////        env.setStateBackend(new HashMapStateBackend());
////        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/flinkCDC/220828");
////        //设置访问HDFS的用户名
////        System.setProperty("HADOOP_USER_NAME", "atguigu");
//
//        //todo 3.创建动态表topic_db连接kafka topic_db
//        tableEnv.executeSql(KafkaUtil.getTopicDbDDL("dwd_trade_payment_detail_suc_220828"));
//
//        //todo 4.过滤出支付表payment_info支付成功的内容
//        Table paymentSucTable = tableEnv.sqlQuery("select\n" +
//                "data['order_id'] order_id, \n" +
//                "data['user_id'] user_id, \n" +
//                "data['payment_type'] payment_type, \n" +
//                "data['callback_time'] callback_time, \n" +
//                "pt\n," +
//                "rt " +
//                "from topic_db\n" +
//                "where `database`='gmall'\n" +
//                "and `table`='payment_info'\n" +
//                "and type='update'\n" +
//                "and old['payment_status'] is not null\n" +
//                "and data['payment_status']='1602'");
//
//        tableEnv.createTemporaryView("payment_suc",paymentSucTable);
//
//        //todo 5.创建动态表连接kafka的下单事务事实表dwd_trade_order_detail
//        tableEnv.executeSql(KafkaUtil.getDwdTradeOrderDetailDDL("dwd_trade_order_detail_220828"));
//
//
//        //todo 6.创建动态表base_dic读取mysql base_dic表
//        tableEnv.executeSql(MysqlUtil.getBaseDicDDL());
//
//        //todo 7.三表关联
//        //先下单再支付，所以下单事务事实表要写到左边（参照官网interval join，时间早的写在左边）
//        tableEnv.sqlQuery("");
//
//        //todo 8.创建动态表trade_order_detail连接kafka dwd_trade_order_detail主题
//        tableEnv.executeSql("")
//
//
//
//        //todo 8.创建动态表payment_detail_suc连接kafka dwd_trade_payment_detail_suc
//        //todo 9.将关联后的结果查询出来写到动态表payment_detail_suc里
//
//
//
//
//    }
//}
